#include "aof_persist.h"
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <errno.h>
#include <time.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include "rbtree.h"
#include "hash.h"
#include "skiplist.h"

char *operate[] =
    {
        "RSET", // 插入操作
        "RMOD", // 更新操作
        "RDEL", // 删除操作
        "RDELA",

        "HSET", // 插入操作
        "HMOD", // 更新操作
        "HDEL", // 删除操作
        "HDELA",

        "SSET", // 插入操作
        "SMOD", // 更新操作
        "SDEL", // 删除操作
        "SDELA",
};

typedef enum
{
    LOG_START = 0,
    LOG_RSET = LOG_START, // 插入操作
    LOG_RMOD,
    LOG_RDEL,
    LOG_RDELA,

    LOG_HSET, // 插入操作
    LOG_HMOD,
    LOG_HDEL,
    LOG_HDELA,

    LOG_SSET, // 插入操作
    LOG_SMOD,
    LOG_SDEL,
    LOG_SDELA,

    LOG_END
} WALOpType;


// ================= 增量持久化 ============================
static WALOpType str_to_op(const char *op_str)
{
    if (!op_str)
        return LOG_END; // 无效字符串

    // 遍历operate数组，找到匹配的字符串（仅一次循环）
    for (WALOpType op = LOG_START; op < LOG_END; op++)
    {
        if (strcmp(op_str, operate[op]) == 0)
        {
            return op; // 匹配成功，返回对应枚举值
        }
    }
    return LOG_END; // 无匹配，返回无效值
}

int is_del(WALOpType op)
{
    return (op == LOG_RDEL || op == LOG_HDEL || op == LOG_SDEL);
}

int is_del_all(WALOpType op)
{
    return (op == LOG_RDELA || op == LOG_HDELA || op == LOG_SDELA);
}

int aof_write(const char *filename, const char *operation, const char *key, const char *value)
{
    if (!filename || !operation)
    {
        fprintf(stderr, "aof_write: 参数无效（NULL）\n");
        return -1;
    }
    // 字符串转枚举（仅一次比较，替代多次strcmp）
    WALOpType op = str_to_op(operation);
    if (op == LOG_END)
    {
        fprintf(stderr, "aof_write: invalid operation '%s'\n", operation);
        return -1;
    }

    // printf("[%s]: Start\n", operation);

    //  通过枚举值索引operate数组（无需再次字符串处理，直接获取）
    const char *op_str = operate[op];
    uint32_t op_len = strlen(op_str);

    // 计算长度（删除操作无值）
    uint32_t key_len = (is_del_all(op)) ? 0 : (key ? strlen(key) : 0);
    uint32_t val_len = (is_del(op) || is_del_all(op)) ? 0 : strlen(value);

    // 长度校验
    if (op_len <= 0)
    {
        fprintf(stderr, "aof_write: invalid length (op_len=%d)\n", op_len);
        return -1;
    }

    // 打开文件

    int fd = 0;
    int flag = 0;
    if (global_aof_writer && strcmp(filename, global_aof_writer->filename) == 0)
    {
        // 判断是否是异步写入在调用
        fd = global_aof_writer->fd;
        flag = 1;
    }
    else
    {
        fd = open(filename, O_WRONLY | O_CREAT | O_APPEND, 0644);
    }

    if (fd == -1)
    {
        perror("aof_write: open failed");
        return -1;
    }

#if 1 // MASTER_SLAVE 为了方便各种模式都能使用正常同一个文件，这里统一存储/读取方式
    // 关键：转换为网络字节序
    uint32_t net_op_len = htonl(op_len);
    uint32_t net_key_len = htonl(key_len);
    uint32_t net_val_len = htonl(val_len);
    struct iovec iov[6];
    int iov_count = 0;
    iov[iov_count++] = (struct iovec){&net_op_len, 4}; // 转换后
    iov[iov_count++] = (struct iovec){(void *)op_str, op_len};
    iov[iov_count++] = (struct iovec){&net_key_len, 4}; // 转换后
    iov[iov_count++] = (struct iovec){&net_val_len, 4}; // 转换后
#else
    // writev分散写入（与全量持久化风格一致）
    struct iovec iov[6];
    int iov_count = 0;
    iov[iov_count++] = (struct iovec){&op_len, 4};             // 操作长度
    iov[iov_count++] = (struct iovec){(void *)op_str, op_len}; // 操作字符串（枚举索引获取）
    iov[iov_count++] = (struct iovec){&key_len, 4};            // 键长度
    iov[iov_count++] = (struct iovec){&val_len, 4};            // 值长度
#endif

    if (key_len > 0)
    {
        iov[iov_count++] = (struct iovec){(void *)key, key_len}; // 键内容
    }

    if (val_len > 0)
    {
        iov[iov_count++] = (struct iovec){(void *)value, val_len}; // 值内容
    }

    // 写入校验
    size_t total_len = 4 + op_len + 4 + 4 + key_len + val_len;
    ssize_t written = writev(fd, iov, iov_count);
    if (written == -1 || (size_t)written != total_len)
    {
        perror("aof_write: writev failed");
        close(fd);
        return -1;
    }

    // printf("[%s]: SUCCESS\n", operation);

    if (flag == 1)
    {
        // 如果是异步写入在调用，直接返回
        return total_len;
    }

    LOG("aof_write: fsync\n");
    fsync(fd); // 刷盘最耗时间了
    close(fd);

    return total_len;
}

int aof_replay(const char *filename, off_t start_offset, RBTree *rbtree, HashTable *hash_table, SkipList *skiplist)
{
    if (!filename)
    {
        fprintf(stderr, "aof_replay: invalid parameters (NULL)\n");
        return -1;
    }

    int fd = open(filename, O_RDONLY);
    if (fd == -1)
    {
        perror("aof_replay: open failed");
        return -1;
    }

    // 设置文件读取的起始偏移量
    off_t current_offset = lseek(fd, start_offset, SEEK_SET);
    if (current_offset == -1)
    {
        perror("aof_replay: lseek to start offset failed");
        close(fd);
        return -1;
    }

    int ret = 0;
    long int aof_read_num = 0;

    while (1)
    {
#if 1 // MASTER_SLAVE 为了方便各种模式都能使用正常同一个文件，这里统一存储/读取方式
        // 1. 读取操作长度（4字节，网络字节序）
        uint32_t net_op_len; // 改为uint32_t，避免符号问题
        ssize_t n = read(fd, &net_op_len, 4);
        if (n == 0)
        { // 正常读取到文件末尾
            ret = 0;
            break;
        }
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read op_len failed\n");
            ret = -1;
            break;
        }
        uint32_t op_len = ntohl(net_op_len); // 关键：网络字节序 → 主机字节序
#else
        // 1. 读取操作长度（4字节）
        uint32_t op_len;
        ssize_t n = read(fd, &op_len, 4);
        if (n == 0)
        { // 正常读取到文件末尾
            ret = 0;
            break;
        }
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read op_len failed\n");
            ret = -1;
            break;
        }
#endif
        // 2. 读取操作字符串
        // if (op_len <= 0 || op_len > 16)
        if (op_len <= 0)
        {
            fprintf(stderr, "aof_replay: invalid op_len %d\n", op_len);
            ret = -1;
            break;
        }
        char *op_str = (char *)kvs_malloc(op_len + 1);
        if (!op_str)
        {
            perror("aof_replay: kvs_malloc op_str failed");
            ret = -1;
            break;
        }
        n = read(fd, op_str, op_len);
        if (n != op_len)
        {
            fprintf(stderr, "aof_replay: read op_str failed\n");
            kvs_free(op_str);
            ret = -1;
            break;
        }
        op_str[op_len] = '\0';

        // 3. 字符串转枚举（仅一次比较，替代后续多次strcmp）
        WALOpType op = str_to_op(op_str);
        if (op == LOG_END)
        {
#if 1 // MASTER_SLAVE 为了方便各种模式都能使用正常同一个文件，这里统一存储/读取方式
            fprintf(stderr, "aof_replay: unknown operation '%s' (skip)\n", op_str);
            kvs_free(op_str);
            // 跳过剩余字段：注意这里也要先读取网络字节序，再转换后跳过
            uint32_t net_key_len, net_val_len;
            if (read(fd, &net_key_len, 4) != 4 || read(fd, &net_val_len, 4) != 4)
            {
                ret = -1;
                break;
            }
            uint32_t key_len_skip = ntohl(net_key_len);
            uint32_t val_len_skip = ntohl(net_val_len);
            lseek(fd, key_len_skip + val_len_skip, SEEK_CUR);
            continue;
#else
            fprintf(stderr, "aof_replay: unknown operation '%s' (skip)\n", op_str);
            kvs_free(op_str);
            // 跳过当前条目的剩余字段（避免影响后续解析）
            uint32_t key_len, val_len;
            read(fd, &key_len, 4);                  // 跳过键长度
            read(fd, &val_len, 4);                  // 跳过值长度
            lseek(fd, key_len + val_len, SEEK_CUR); // 跳过键和值
            continue;
#endif
        }

#if 1 // MASTER_SLAVE 为了方便各种模式都能使用正常同一个文件，这里统一存储/读取方式
        // 4. 读取键长度（4字节，网络字节序）
        uint32_t net_key_len;
        n = read(fd, &net_key_len, 4);
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read key_len failed\n");
            kvs_free(op_str);
            ret = -1;
            break;
        }
        uint32_t key_len = ntohl(net_key_len); // 关键：转换字节序

        // if (key_len < 0 || key_len > 4096)
        if (key_len < 0)
        {
            fprintf(stderr, "aof_replay: invalid key_len %d\n", key_len);
            kvs_free(op_str);
            ret = -1;
            break;
        }

        // 5. 读取值长度（4字节，网络字节序）
        uint32_t net_val_len;
        n = read(fd, &net_val_len, 4);
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read val_len failed\n");
            kvs_free(op_str);
            ret = -1;
            break;
        }
        uint32_t val_len = ntohl(net_val_len); // 关键：转换字节序

        // if (val_len < 0 || val_len > 4096)
        if (val_len < 0)
        {
            fprintf(stderr, "aof_replay: invalid val_len %d\n", val_len);
            kvs_free(op_str);
            ret = -1;
            break;
        }
#else
        // 4. 读取键长度
        uint32_t key_len;
        n = read(fd, &key_len, 4);
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read key_len failed\n");
            kvs_free(op_str);
            ret = -1;
            break;
        }

        // if (key_len < 0 || key_len > 4096)
        if (key_len < 0)
        {
            fprintf(stderr, "aof_replay: invalid key_len %d\n", key_len);
            kvs_free(op_str);
            ret = -1;
            break;
        }

        // 5. 读取值长度
        uint32_t val_len;
        n = read(fd, &val_len, 4);
        if (n != 4)
        {
            fprintf(stderr, "aof_replay: read val_len failed\n");
            kvs_free(op_str);
            ret = -1;
            break;
        }
        // if (val_len < 0 || val_len > 4096)
        if (val_len < 0)
        {
            fprintf(stderr, "aof_replay: invalid val_len %d\n", val_len);
            kvs_free(op_str);
            ret = -1;
            break;
        }
#endif
        // 6. 读取键
        char *key = NULL;
        if (key_len > 0)
        {
            key = (char *)kvs_malloc(key_len + 1);
            if (!key)
            {
                perror("aof_replay: 分配键内存失败");
                kvs_free(op_str);
                ret = -1;
                break;
            }
            n = read(fd, key, key_len);
            if (n != key_len)
            {
                fprintf(stderr, "aof_replay: 读取键失败\n");
                kvs_free(op_str);
                kvs_free(key);
                ret = -1;
                break;
            }
            key[key_len] = '\0';
        }

        // 7. 读取值
        char *value = NULL;
        if (val_len > 0)
        {
            value = (char *)kvs_malloc(val_len + 1);
            if (!value)
            {
                perror("aof_replay: kvs_malloc value failed");
                kvs_free(op_str);
                kvs_free(key);
                ret = -1;
                break;
            }
            n = read(fd, value, val_len);
            if (n != val_len)
            {
                fprintf(stderr, "aof_replay: read value failed\n");
                kvs_free(op_str);
                kvs_free(key);
                kvs_free(value);
                ret = -1;
                break;
            }
            value[val_len] = '\0';
        }

        // 8. 用枚举值判断操作（替代多次strcmp，效率更高）

        switch (op)
        {
        // rbtree
        case LOG_RSET:
            rb_insert(rbtree, key, value);
            break;
        case LOG_RMOD:
            rb_update(rbtree, key, value);
            break;
        case LOG_RDEL:
            rb_delete(rbtree, key);
            break;
        case LOG_RDELA:
            rb_clear(rbtree);
            break;

        // hash
        case LOG_HSET:
            hashtable_insert(hash_table, key, value);
            break;
        case LOG_HMOD:
            hashtable_update(hash_table, key, value);
            break;
        case LOG_HDEL:
            hashtable_delete(hash_table, key);
            break;
        case LOG_HDELA:
            hashtable_clear(hash_table);
            break;

        // skiplist
        case LOG_SSET:
            skiplist_insert(skiplist, key, value);
            break;
        case LOG_SMOD:
            skiplist_update(skiplist, key, value);
            break;
        case LOG_SDEL:
            skiplist_delete(skiplist, key);
            break;
        case LOG_SDELA:
            skiplist_clear(skiplist);
            break;

        default: // 理论上不会走到这里（已在str_to_op中过滤）
            break;
        }

        aof_read_num++;
        // printf("replay: 操作=%s, 键=%s, 值=%s\n", op_str, key ? key : "NULL", value ? value : "NULL");

        // 释放内存
        kvs_free(op_str);
        if (key)
            kvs_free(key);
        if (value)
            kvs_free(value);
    }

    LOG("成功读取了%ld条日志\n", aof_read_num);

    close(fd);
    return ret;
}

// ================ 增量的异步更新 ================

// AOF日志条目结构体（去掉const，明确由异步模块管理内存）
typedef struct
{
    char *filename; // 可释放的字符串
    char *operation;
    char *key;
    char *value;
} AOFLogEntry;

// 辅助函数：计算两个timespec的时间差（毫秒）
static unsigned long timespec_diff_ms(const struct timespec *start, const struct timespec *end)
{
    long sec_diff = end->tv_sec - start->tv_sec;
    long nsec_diff = end->tv_nsec - start->tv_nsec;
    // 处理纳秒溢出（如end->tv_nsec < start->tv_nsec）
    if (nsec_diff < 0)
    {
        sec_diff--;
        nsec_diff += 1000000000; // 1秒 = 1e9纳秒
    }
    // 秒转毫秒 + 纳秒转毫秒
    return (unsigned long)(sec_diff * 1000) + (unsigned long)(nsec_diff / 1000000);
}

// 后台线程函数（接收AOFAsyncWriter实例指针）
static void *aof_worker_thread(void *arg)
{
    AOFAsyncWriter *writer = (AOFAsyncWriter *)arg;
    if (!writer)
        return NULL;

    struct timespec now;

    off_t write_total_num = 0;
    while (atomic_load(&writer->running) || atomic_load(&writer->queue->task_count) > 0)
    {
        // 先尝试消费任务
        AOFLogEntry *entry = (AOFLogEntry *)spsc_list_queue_dequeue(writer->queue);
        if (entry)
        {
            // 调用aof_write写入日志
            int ret = aof_write(
                entry->filename ? entry->filename : writer->filename,
                entry->operation,
                entry->key,
                entry->value);

            if (ret >= 0)
            {
#if MASTER
                write_total_num += ret;
                // atomic_fetch_add(&aof_file_sync_length, ret);
#endif

                writer->has_unflushed = true;
            }
            else
            {
                fprintf(stderr, "aof_worker_thread: aof_write failed\n");
            }

            // 释放日志条目内存（确保每个任务仅释放一次）
            kvs_free(entry->filename);
            kvs_free(entry->operation);
            kvs_free(entry->key);
            kvs_free(entry->value);
            kvs_free(entry);
        }
        else
        {
            usleep(1000); // 1ms，减少CPU占用
        }

        if (writer->has_unflushed)
        {

            if (writer->flush_interval == 0) // 间隔时间为0， 每次都刷盘
            {
                if (fsync(writer->fd) == -1)
                {
                    perror("aof_worker_thread: fsync failed");
                }
                else
                {
                    writer->has_unflushed = false;
                    printf("aof_worker_thread: 刷盘成功, 间隔: 0ms\n");
#if MASTER
                    atomic_fetch_add(&aof_file_sync_length, write_total_num);
                    write_total_num = 0;
#endif
                }
                continue;
            }

            // 2. 定时检查：是否需要刷盘
            if (clock_gettime(CLOCK_MONOTONIC, &now) == -1)
            {
                perror("aof_worker_thread: clock_gettime failed");
                continue;
            }

            // 计算距离上次刷盘的时间差
            unsigned long diff_ms = timespec_diff_ms(&writer->last_flush, &now);
            // 到达刷盘间隔 + 有未刷盘数据 → 执行刷盘
            if (diff_ms >= writer->flush_interval)
            {
                if (fsync(writer->fd) == -1)
                {
                    perror("aof_worker_thread: fsync failed");
                }
                else
                {
                    // 刷盘成功：更新上次刷盘时间 + 重置未刷盘标志
                    writer->last_flush = now;
                    writer->has_unflushed = false;
                    printf("aof_worker_thread: 定时刷盘成功（间隔：%lu ms）\n", diff_ms);

#if MASTER
                    atomic_fetch_add(&aof_file_sync_length, write_total_num);
                    write_total_num = 0;
#endif
                }
            }
        }
    }

    // 3. 线程退出前：强制刷盘一次（确保所有数据落盘）
    if (writer->has_unflushed)
    {
        // printf("aof_worker_thread: 剩余任务处理完毕... 强制刷盘\n");
        if (fsync(writer->fd) == -1)
        {
            perror("aof_worker_thread: exit fsync failed");
        }
    }
    printf("aof_worker_thread: 剩余任务处理完毕... 强制刷盘\n");

    return NULL;
}

// 初始化异步写入器（创建队列、启动后台线程）

AOFAsyncWriter *aof_async_writer_init(const char *filename, unsigned int flush_ms)
{
    if (!filename)
        return NULL;

    AOFAsyncWriter *writer = (AOFAsyncWriter *)kvs_malloc(sizeof(AOFAsyncWriter));
    if (!writer)
        return NULL;

    writer->filename = kvs_malloc(strlen(filename) + 1);
    memset(writer->filename, 0, strlen(filename) + 1);
    strcpy(writer->filename, filename);

    // 打开文件（追加模式，不存在则创建，权限0644）
    writer->fd = open(writer->filename, O_WRONLY | O_CREAT | O_APPEND, 0644);
    if (writer->fd == -1)
    {
        perror("aof_async_writer_init: open failed");
        kvs_free(writer->filename);
        kvs_free(writer);
        return NULL;
    }

    // 初始化无锁队列
    writer->queue = spsc_list_queue_init();
    if (!writer->queue)
    {
        kvs_free(writer->filename);

        close(writer->fd);
        kvs_free(writer);

        return NULL;
    }

    // 初始化原子变量（C11 atomic_store 保证原子性）
    atomic_store(&writer->running, true);

    writer->has_unflushed = false;
    // 初始化定时参数（最小刷盘间隔10ms，默认3秒）
    if (flush_ms == 0)
    {
        writer->flush_interval = 0;
    }
    else
    {
        writer->flush_interval = (flush_ms > 10) ? flush_ms : 3000;
    }

    // 使用单调时间（CLOCK_MONOTONIC），不受系统时间调整影响
    if (clock_gettime(CLOCK_MONOTONIC, &writer->last_flush) == -1)
    {
        perror("aof_async_writer_init: clock_gettime failed");
        spsc_list_queue_destroy(writer->queue);
        kvs_free(writer->filename);

        close(writer->fd);
        kvs_free(writer);
        return NULL;
    }

    // 启动后台线程（线程函数接收writer指针）
    int ret = pthread_create(&writer->thread_id, NULL, aof_worker_thread, writer);
    if (ret != 0)
    {
        spsc_list_queue_destroy(writer->queue);

        kvs_free(writer->filename);

        close(writer->fd);
        kvs_free(writer);
        return NULL;
    }

    return writer;
}

// 销毁异步写入器（等待线程、销毁队列）
void aof_async_writer_destroy(AOFAsyncWriter *writer)
{
    if (!writer)
        return;

    // 通知线程退出
    atomic_store(&writer->running, false);

    printf("线程退出，等待后台线程剩余任务\n");

    // 等待队列任务处理完毕（循环检查任务计数）
    while (atomic_load(&writer->queue->task_count) > 0)
    {
        // usleep(10000); // 10ms休眠，避免忙等
        sleep(5);
        // printf("等待后台线程处理剩余任务... 剩余任务数：%d\n", atomic_load(&writer->queue->task_count));
    }

    // 等待线程结束
    pthread_join(writer->thread_id, NULL);

    // 销毁队列
    spsc_list_queue_destroy(writer->queue);
    kvs_free(writer->filename);

    close(writer->fd);
    kvs_free(writer);
}

// 安全复制字符串（失败时返回NULL，释放时兼容NULL）
static char *str_dup_safe(const char *src)
{
    if (!src)
        return NULL;
    char *dst = strdup(src);
    if (!dst)
        perror("str_dup_safe: strdup failed");
    return dst;
}

// ===================== 主线程 ========================

// 异步写入函数（接收AOFAsyncWriter实例指针）
int aof_async_write(AOFAsyncWriter *writer, const char *filename, const char *operation, const char *key, const char *value)
{
    if (!writer || !operation)
    {
        fprintf(stderr, "aof_async_write: invalid parameters\n");
        return -1;
    }

    // 创建日志条目并复制字符串
    AOFLogEntry *entry = (AOFLogEntry *)kvs_malloc(sizeof(AOFLogEntry));
    if (!entry)
    {
        perror("aof_async_write: kvs_malloc entry failed");
        return -1;
    }
    entry->filename = str_dup_safe(filename);
    entry->operation = str_dup_safe(operation);
    entry->key = str_dup_safe(key);
    entry->value = str_dup_safe(value);

    // 校验复制结果
    if (!entry->operation)
    {
        kvs_free(entry->filename);
        kvs_free(entry->operation);
        kvs_free(entry->key);
        kvs_free(entry->value);
        kvs_free(entry);
        return -1;
    }

    // 入队（操作实例的队列）
    if (spsc_list_queue_enqueue(writer->queue, entry) != 0)
    {
        kvs_free(entry->filename);
        kvs_free(entry->operation);
        kvs_free(entry->key);
        kvs_free(entry->value); // kvs_free NULL不会报错
        kvs_free(entry);
        return -1;
    }

    return 0;
}
